package com.chinaums.sftp;

import com.chinaums.sftp.internal.options.SftpReadOptions;
import com.chinaums.sftp.tablesource.SftpDynamicTableSource;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DeserializationFormatFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;
import org.apache.flink.util.Preconditions;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;

public class SftpDynamicTableFactory implements DynamicTableSourceFactory {
    public static final String IDENTIFIER = "sftp";

    public static final ConfigOption<String> HOSTNAME = ConfigOptions.key("hostname")
            .stringType()
            .noDefaultValue()
            .withDescription("hostname...");
    public static final ConfigOption<Integer> PORT = ConfigOptions.key("port")
            .intType()
            .defaultValue(22)
            .withDescription("端口...");
    public static final ConfigOption<String> USERNAME = ConfigOptions.key("username")
            .stringType()
            .noDefaultValue()
            .withDescription("用户名");
    public static final ConfigOption<String> PASSWORD = ConfigOptions.key("password")
            .stringType()
            .noDefaultValue()
            .withDescription("密码...");
    public static final ConfigOption<String> PATH = ConfigOptions.key("path")
            .stringType()
            .noDefaultValue()
            .withDescription("文件路径...");
    public static final ConfigOption<String> ENCODING = ConfigOptions.key("encoding")
            .stringType()
            .defaultValue("UTF-8")
            .withDescription("编码...");
    public static final ConfigOption<Boolean> IS_FIRST_LINE_HEADER = ConfigOptions
            .key("first-line-header")
            .booleanType()
            .defaultValue(false)
            .withDescription("是否跳过文件首行...");
    public static final ConfigOption<Integer> TIME_OUT = ConfigOptions.key("timeout")
            .intType()
            .defaultValue(10000)
            .withDescription("sftp默认超时时间...");
    public static final ConfigOption<String> READ_MODE = ConfigOptions
            .key("read-mode")
            .stringType()
            .defaultValue("STREAM");

    // 将已读文件删除
    public static final ConfigOption<Boolean> DELETE_READ = ConfigOptions
            .key("delete-read")
            .booleanType()
            .defaultValue(false);
    public static final ConfigOption<Integer> BYTE_DELIMITER_LENGTH = ConfigOptions
            .key("byte-delimiter-length")
            .intType()
            .defaultValue(1);

    public static final ConfigOption<Boolean> FTP_CHECK_OK_FILE = ConfigOptions
            .key("check-ok-file")
            .booleanType()
            .defaultValue(false)
            .withDescription(" 检测ok文件 ");

    public static final ConfigOption<Boolean> FTP_IS_METRIC = ConfigOptions
            .key("is-metric")
            .booleanType()
            .defaultValue(false)
            .withDescription(" 是否开启监控，监控数据会写入到mysql表中 ");


    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);

        final DecodingFormat<DeserializationSchema<RowData>> decodingFormat = helper.discoverDecodingFormat(
                DeserializationFormatFactory.class,
                FactoryUtil.FORMAT);

        helper.validate();

        final ReadableConfig options = helper.getOptions();

        ConfigOption<?>[] configOptions = {USERNAME, PASSWORD};
        int presentCount = 0;
        for (ConfigOption<?> configOption : configOptions) {
            if (options.getOptional(configOption).isPresent()) {
                presentCount++;
            }
        }
        Preconditions.checkArgument(
                configOptions.length == presentCount || presentCount == 0,
                "Either all or none of the following options should be provided:\n" + String.join(
                        "\n",
                        getPropertyNames(configOptions)));

        // 1.13 新的 PhysicalRowDataType
        final DataType producedDataType =
                context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType();

        return new SftpDynamicTableSource(
                getFtpReadOptions(options),
                decodingFormat,
                producedDataType);
    }

    @Override
    public String factoryIdentifier() {
        return IDENTIFIER;
    }

    /**
     * 必须传入的参数
     */
    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        Set<ConfigOption<?>> requiredOptions = new HashSet<>();
        requiredOptions.add(HOSTNAME);
        requiredOptions.add(USERNAME);
        requiredOptions.add(PASSWORD);
        requiredOptions.add(PATH);
        requiredOptions.add(FactoryUtil.FORMAT);
        return requiredOptions;
    }

    /**
     * 选添的参数
     */
    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        Set<ConfigOption<?>> optionalOptions = new HashSet<>();
        optionalOptions.add(FactoryUtil.FORMAT);
        optionalOptions.add(PORT);
        optionalOptions.add(USERNAME);
        optionalOptions.add(PASSWORD);
        optionalOptions.add(ENCODING);
        optionalOptions.add(IS_FIRST_LINE_HEADER);
        optionalOptions.add(TIME_OUT);
        optionalOptions.add(READ_MODE);
        optionalOptions.add(DELETE_READ);
        optionalOptions.add(BYTE_DELIMITER_LENGTH);
        optionalOptions.add(FTP_CHECK_OK_FILE);
        optionalOptions.add(FTP_IS_METRIC);
        return optionalOptions;
    }

    /**
     * 参数封装
     */
    private SftpReadOptions getFtpReadOptions(ReadableConfig config) {
        return SftpReadOptions.builder()
                .setHost(config.get(HOSTNAME))
                .setPort(config.get(PORT))
                .setUsername(config.get(USERNAME))
                .setPassword(config.get(PASSWORD))
                .setPath(config.get(PATH))
                .setEncoding(config.get(ENCODING))
                .setIsFirstLineHeader(config.get(IS_FIRST_LINE_HEADER))
                .setTimeout(config.get(TIME_OUT))
                .setReadMode(config.get(READ_MODE))
                .setDeleteRead(config.get(DELETE_READ))
                .setByteDelimiter(config.get(BYTE_DELIMITER_LENGTH))
                .setFileCheckOkFile(config.get(FTP_CHECK_OK_FILE))
                .setMetric(config.get(FTP_IS_METRIC))
                .build();
    }

    private String[] getPropertyNames(ConfigOption<?>[] configOptions) {
        return Arrays
                .stream(configOptions)
                .map(ConfigOption::key)
                .toArray(String[]::new);
    }
}
